package com.hkbigdata.streamCoreCoding;

import com.hkbigdata.entry.UserBehavior;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashSet;


/**
 * @author liuanbo
 * @creat 2023-04-10-13:46
 * @see 2194550857@qq.com
 *
 *
 */
public class Flink03_UserVisitor {
    public static void main(String[] args) throws Exception {
        //1.运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.读取数据
        DataStreamSource<String> source = env.readTextFile("input/UserBehavior.csv");

        //3.拍平数据
        SingleOutputStreamOperator<UserBehavior> userBehaviorSingleOutputStreamOperator = source.flatMap(new FlatMapFunction<String, UserBehavior>() {
            @Override
            public void flatMap(String value, Collector<UserBehavior> out) throws Exception {
                String[] split = value.split(",");
                UserBehavior userBehavior = new UserBehavior(
                        Long.parseLong(split[0]),
                        Long.parseLong(split[1]),
                        Integer.parseInt(split[2]),
                        split[3],
                        Long.parseLong(split[4])
                );

                if ("pv".equals(userBehavior.getBehavior())) {
                    out.collect(userBehavior);
                }
            }
        });

        //4.分组
        KeyedStream<UserBehavior, String> userBehaviorStringKeyedStream = userBehaviorSingleOutputStreamOperator.keyBy(data -> "UV");

        //5.聚合
        SingleOutputStreamOperator<Integer> process = userBehaviorStringKeyedStream.process(new KeyedProcessFunction<String, UserBehavior, Integer>() {
            private Integer count = 0;

            private HashSet uids = new HashSet();

            @Override
            public void processElement(UserBehavior value, Context ctx, Collector<Integer> out) throws Exception {
                //所谓的重复就说存在一个用户多次刷新网页的情况，所以按照用户id进行去重
                if (!uids.contains(value.getUserId())) {
                    //如果不在集合里面，那么说明是第一次刷新网页，那么就保存到集合中，同时计数加一
                    uids.add(value.getUserId());

                    count++;
                    out.collect(count);
                }
            }
        });
        //动作
        process.print();

        env.execute();
    }
}
